package com.lzhsite.mapreduce.sort;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class SecondSortMapReduce {

	/**
	 * 使用内部类的形式，定义mapper程序
	 * 
	 * @author Administrator 2018年5月31日上午11:06:30
	 */
	static class SecondSortMapper extends Mapper<LongWritable, Text, CombinationKey, IntWritable> {
		String[] split = null;
		CombinationKey kv = new CombinationKey();
		IntWritable v = new IntWritable();

		@Override
		protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
			split = value.toString().split(" ");
			kv.setFirstKey(split[0]);
			int vv = Integer.parseInt(split[1]);
			v.set(vv);
			kv.setSecondKey(vv);
			context.write(kv, v);
		}
	}

	/**
	 * 使用内部类的形式，定义reduce程序
	 * 
	 * @author Administrator 2018年5月31日上午11:06:51
	 */
	static class SecondSortReducer extends Reducer<CombinationKey, IntWritable, Text, Text> {
		Text k = new Text();
		Text v = new Text();

		// reducer类的操作对象是分区，一个分区初始化一次reducer类。
		// reduce()函数的操作对象是组，一个分区中有几个分组就调用几次reduce()函数
		@Override
		protected void reduce(CombinationKey first_second, Iterable<IntWritable> seconds, Context context)
				throws IOException, InterruptedException {
			StringBuilder sb = new StringBuilder();
			for (IntWritable second : seconds) {
				sb.append(second.get() + ",");
			}
			k.set(first_second.getFirstKey());
			v.set(sb.toString().substring(0, sb.toString().length() - 1));
			context.write(k, v);
		}
	}

	public int run(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

		Configuration config = new Configuration();
		Job job = Job.getInstance(config, SecondSortMapReduce.class.getSimpleName());

		job.setJarByClass(SecondSortMapReduce.class);
		job.setMapperClass(SecondSortMapper.class);
		job.setReducerClass(SecondSortReducer.class);

		// 设置分区和reduce数目
		job.setPartitionerClass(DefinedPartition.class);
		job.setNumReduceTasks(1);

		// 设置自定义的排序策略
		job.setSortComparatorClass(DefineCompparator.class);
		// 设置自定义的分组策略
		job.setGroupingComparatorClass(DefinedGroupSort.class);
		
		job.setMapOutputKeyClass(CombinationKey.class);
		job.setMapOutputValueClass(IntWritable.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		// 设置输出数据
		Path outPath = new Path(args[1]);
		FileInputFormat.setInputPaths(job, new Path(args[0]));
		FileSystem fsh = outPath.getFileSystem(config);
		if (fsh.exists(outPath)) {
			fsh.delete(outPath, true);
		}
		FileOutputFormat.setOutputPath(job,outPath);

		boolean isSuccess = job.waitForCompletion(true);
		return isSuccess ? 0 : 1;
	}

	/**
	 * 主函数
	 * 
	 * @param args
	 * @throws IOException
	 * @throws ClassNotFoundException
	 * @throws InterruptedException
	 */
	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

		// 定义输入路径
		String INPUT_PATH = "file:///"
				+ Thread.currentThread().getContextClassLoader().getResource("data/sort/sort.txt").getPath();
		// 定义输出路径
		String OUT_PATH = "hdfs://aliyun.lzh/user/beifeng/sort";
		args = new String[] { INPUT_PATH, OUT_PATH };

		int status = new SecondSortMapReduce().run(args);

		System.exit(status);

	}
}
